package com.bw.flinkstreaming.state.job3;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.shaded.guava18.com.google.common.collect.Lists;
import org.apache.flink.util.Collector;
import java.util.List;
import java.util.UUID;
/**
 *  MapState<K, V> ：这个状态为每一个 key 保存一个 Map 集合
 *  put() 将对应的 key 的键值对放到状态中
 *  values() 拿到 MapState 中所有的 value
 *  clear() 清除状态
 */
public class MapStateWithCountAvg extends RichFlatMapFunction<Tuple2<Long, Long>, Tuple2<Long, Double>> {

    /**
     * 每个key 都有自己的一个state，相同的key都会有一个自己的state
     * key相同的数据会覆盖？
     */
    private MapState<String, Long> mapState;

    @Override
    public void open(Configuration parameters) throws Exception {
        // 注册状态
        MapStateDescriptor<String, Long> descriptor = new MapStateDescriptor<String, Long>(
                        // 状态的名字
                        "average",
                        // 状态存储的数据类型
                        String.class, Long.class);
        mapState = getRuntimeContext().getMapState(descriptor);
    }

    /**
     * element：接收到的参数
     * out：输出对象
     */
    @Override
    public void flatMap(Tuple2<Long, Long> element, Collector<Tuple2<Long, Double>> out) throws Exception {
        mapState.put(UUID.randomUUID().toString(), element.f1);
        // 判断，如果当前的 key 出现了 3 次，则需要计算平均值，并且输出
        List<Long> allElements = Lists.newArrayList(mapState.values());

        if (allElements.size() == 3) {
            long count = 0;
            long sum = 0;
            for (Long ele : allElements) {
                count++;
                sum += ele;
            }
            double avg = (double) sum / count;
            out.collect(Tuple2.of(element.f0, avg));
            // 清除状态
            mapState.clear();
        }
    }
}
